package com.zhang.sql2;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

/**
 * @title:
 * @author: zhang
 * @date: 2022/3/19 16:43
 */
public class CommonApiTest {
    public static void main(String[] args) {
/*        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);*/

        //1 .定义环境配置来创建表执行环境 新版本基于blink的流处理
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();

        TableEnvironment tableEnvironment = TableEnvironment.create(settings);

        //2 创建表 (连接器表)
        String create = "create table clickTable (" +
                " username STRING," +
                " url STRING," +
                " ts BIGINT" +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'input/click.txt'," +
                " 'format' = 'csv'" +
                " )";
        tableEnvironment.executeSql(create);

        //3 创建输出表
        String createOut = "create table outTable (" +
                " username STRING," +
                " url STRING " +
                " ) WITH ( " +
                " 'connector' = 'filesystem'," +
                " 'path' = 'output'," +
                " 'format' = 'csv'" +
                " )";
        tableEnvironment.executeSql(createOut);


        //4 创建控制台打印输出表
        String printOut = "create table printTable (" +
                " username STRING," +
                " cnt BIGINT " +
                " ) WITH ( " +
                " 'connector' = 'print'" +
                " )";
        tableEnvironment.executeSql(printOut);

        //4.查询计算
        Table resultTable = tableEnvironment.sqlQuery("select username,url from clickTable");
        //聚合计算
        Table aggResult = tableEnvironment.sqlQuery("" +
                " select username,count(url) cnt " +
                " from clickTable " +
                " group by username");

        //resultTable.executeInsert("printTable");
        aggResult.executeInsert("printTable");


    }
}
